package com.tensai.study.framework.reactornetty;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.ssl.util.SelfSignedCertificate;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.netty.Connection;
import reactor.netty.tcp.TcpServer;

import java.security.cert.CertificateException;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author Tensai
 */
@Slf4j
public class MyTcpServerTest {
    Map<String, MyServerConnection> myConnMap = new ConcurrentHashMap<>();

    @Test
    public void main() throws CertificateException {
        SelfSignedCertificate cert = new SelfSignedCertificate();
        Flux.interval(Duration.ofSeconds(5))
            .subscribe(it -> {
                List<String> list = new ArrayList<>();
                log.info("{} -- 当前链接：{} 个", it, myConnMap.size());
                myConnMap.forEach((k, v) -> {
                    if (v.conn.isDisposed()) {
//                        log.info("{} -- 失活链接：{}", it, k);
                        list.add(k);
                    } else {
//                        log.info("{} -- 活动链接：{}", it, k);
                    }
                });
                log.info("{} -- 失活链接：{} 个", it, list.size());
                list.forEach(k-> {
                    log.info("{} -- 清理链接：{}", it, k);
                    myConnMap.remove(k);
                });
            });

        // 准备要配置的TCP服务器。
        TcpServer.create()
                 // 配置端口号
                 .port(11551)
                 // 使用自签名证书。
//				.secure(spec -> spec.sslContext(TcpSslContextSpec.forServer(cert.certificate(), cert.privateKey())))
                 //.wiretap()
                 .doOnConnection(this::onConnect)
                 // 以阻塞方式启动服务器，并等待其完成初始化。
                 .bindUntilJavaShutdown(Duration.ofSeconds(30), null);

    }

    private void onConnect(Connection conn) {
        MyServerConnection myConn = new MyServerConnection(conn);
        // 将处理程序添加到netty管道
        conn.addHandlerFirst(new StringToIntegerDecoder(myConn));
        log.info("新客户端链接: {}", conn);
        myConn.handle();
        myConnMap.put(conn.address().toString(), myConn);
    }

    /**
     * 返回接收字节数的解码器。
     */
    @AllArgsConstructor
    public static class StringToIntegerDecoder extends ByteToMessageDecoder {

        MyServerConnection myConn;


        @Override
        protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
            int n = in.readableBytes();
            byte[] b = new byte[n];
            if (in.readableBytes() > 0) {
                in.readBytes(b);
                // 将解码器的结果存储在此处
                out.add(new String(b));
                myConn.sinks.tryEmitNext(n + "");
            }
        }
    }
}
